package com.vhall.component.service.interactive.impl;

import cn.hutool.core.text.CharSequenceUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
import com.baomidou.mybatisplus.core.conditions.update.UpdateWrapper;
import com.baomidou.mybatisplus.core.toolkit.CollectionUtils;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.vhall.component.dao.account.AccountMapper;
import com.vhall.component.dao.account.RoleMapper;
import com.vhall.component.dao.interactive.QuestionAnswersMapper;
import com.vhall.component.entity.account.AccountEntity;
import com.vhall.component.entity.account.RoleEntity;
import com.vhall.component.entity.account.dto.IndexDto;
import com.vhall.component.entity.data.enums.SelectTimeIntervalOptionsEnum;
import com.vhall.component.entity.interactive.dto.QuestionAnswerNumberCacheDTO;
import com.vhall.component.entity.interactive.entity.QuestionAnswersEntity;
import com.vhall.component.entity.interactive.entity.QuestionsEntity;
import com.vhall.component.entity.interactive.vo.*;
import com.vhall.component.entity.room.RoomQuestionLkEntity;
import com.vhall.component.entity.room.dto.RoomJoinsEntityDTO;
import com.vhall.component.entity.room.dto.RoomsEntityDTO;
import com.vhall.component.framework.common.constant.CommonConstant;
import com.vhall.component.framework.common.exception.BusinessException;
import com.vhall.component.framework.common.exception.GlobalErrorCode;
import com.vhall.component.plugin.common.constant.RedisKey;
import com.vhall.component.plugin.common.exception.BizErrorCode;
import com.vhall.component.plugin.common.support.VssToken;
import com.vhall.component.plugin.common.support.VssTokenThreadLocal;
import com.vhall.component.plugin.common.utils.JsonUtil;
import com.vhall.component.plugin.common.utils.RedisLocker;
import com.vhall.component.service.interactive.QuestionCacheService;
import com.vhall.component.service.interactive.QuestionService;
import com.vhall.component.service.interactive.assembler.QuestionAssembler;
import com.vhall.component.service.interactive.common.QuestionConsts;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.time.LocalDate;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * @author guoxin.wang
 * @date 2021-06-10 15:21
 */
@Slf4j
@Service
public class QuestionServiceImpl extends QuestionBaseServiceImpl implements QuestionService {

    @Autowired
    private QuestionAnswersMapper answersMapper;

    @Autowired
    private QuestionCacheService questionCacheService;

    @Autowired
    private QuestionAssembler questionAssembler;
    public static final String ROOM_ID = "room_id";
    public static final String QUESTION_ID = "question_id";
    public static final String ACCOUNT_ID = "account_id";
    public static final String LIMIT_1 = "limit 1";
    public static final String JOIN_ID = "join_id";
    public static final String DELETED_AT = "deleted_at";
    public static final String CREATED_AT = "created_at";
    /**
     * 推屏时间间隔
     */
    private static final int QUESTION_RE_PUSH_INTERVAL_TIMEOUT = 10;
    @Autowired
    private AccountMapper accountMapper;
    @Autowired
    private RoleMapper roleMapper;

    @Override
    public QuestionPage<QuestionListRspVO> list(QuestionListQuery req) {
        List<QuestionListRspVO> result = new ArrayList<>();
        String keyword = req.getKeyword();
        Integer offset = (req.getPage() - 1) * req.getPageSize();
        Integer bind = 1;
        List<Integer> accountIds = null;
        if (req.getAccountId() != null){
            RoomsEntityDTO roomInfo = roomInnerService.getRoomInfo(req.getFromRoomId());
            AccountEntity accountEntity = accountMapper.selectById(roomInfo.getAccountId());
            Integer roleId = accountEntity.getRoleId();
            if (roleId != null){
                RoleEntity roleEntity = roleMapper.selectById(roleId);
                if (roleEntity.getType() != null && roleEntity.getType() == 2) {
                    LambdaQueryWrapper<AccountEntity> queryWrapper = Wrappers.lambdaQuery();
                    queryWrapper.eq(AccountEntity::getOrgCode, accountEntity.getOrgCode());
                    List<AccountEntity> accountEntities = accountMapper.selectList(queryWrapper);
                    accountIds = accountEntities.stream().map(AccountEntity::getAccountId).collect(Collectors.toList());
                }
            }

        }

        List<QuestionListDTO> qList = questionRoomMapper.listQuestion(keyword, req.getRoomId(), bind, req.getPublish(),
                accountIds, offset, req.getPageSize());
        Integer count = questionRoomMapper.countQuestion(keyword, req.getRoomId(), bind, req.getPublish(), accountIds);
        if (CollectionUtils.isEmpty(qList)) {
            return new QuestionPage<>(req.getPage(), req.getPageSize(), count, result);
        }
        // 实体拷贝了
        result = qList.stream().map(item -> {
            QuestionListRspVO nvo = new QuestionListRspVO();
            BeanUtils.copyProperties(item, nvo);
            return nvo;
        }).collect(Collectors.toList());
        // 2.1 主持端 传from_room_id, 查询直播间里所有问卷列表
        if (null != req.getFromRoomId() && CollectionUtils.isNotEmpty(result)) {
            fromRoomIdGetQuestionList(req, result);

        }
        // 3、查询问卷的填写人数。
        Map<String, Integer> mapOfCount = new HashMap<>(16);
        Map<String, Integer> mapOfAccount = new HashMap<>(16);
        Map<String, RoomsEntityDTO> mapOfRoomInfo = new HashMap<>(16);
        buildQuestionAccountMap(req, result, mapOfCount, mapOfAccount);
        // 给集合中的ilId赋值。
        buildQuestionRoomInfoMap(result, mapOfRoomInfo);
        // 5、给集合中的数据赋值。
        buildQuestionListRspVOList(result, mapOfCount, mapOfAccount, mapOfRoomInfo);
        return new QuestionPage<>(req.getPage(), req.getPageSize(), count, result);
    }

    /**
     * 构建问卷与用户和填写问卷数量的映射关系
     *
     * @param req          问卷查询参数
     * @param result       问卷数据列表
     * @param mapOfCount   用户填写数量映射
     * @param mapOfAccount 填写人映射
     */
    private void buildQuestionAccountMap(QuestionListQuery req, List<QuestionListRspVO> result, Map<String, Integer> mapOfCount, Map<String, Integer> mapOfAccount) {
        List<QuestionCountVo> accounts;
        List<QuestionCountVo> counts;
        List<Integer> questionIds = result.stream().map(QuestionListRspVO::getQuestionId).collect(Collectors.toList());
        if (CollectionUtils.isNotEmpty(questionIds)) {
            counts = questionAnswersMapper.getAnswerCountByQuestionIds(questionIds);
            for (QuestionCountVo v : counts) {
                mapOfCount.put(String.valueOf(v.getQuestionId()), v.getCount());
            }

            // 4、查询当前用户是否填写了问卷。
            accounts = questionAnswersMapper.getWhetherMap(questionIds, req.getAccountId());
            for (QuestionCountVo v : accounts) {
                mapOfAccount.put(String.valueOf(v.getQuestionId()), v.getAnswerId());
            }
        }
    }

    /**
     * 通过房间信息获取,房间的问卷信息
     *
     * @param req    房间信息
     * @param result 问卷列表
     */
    private void fromRoomIdGetQuestionList(QuestionListQuery req, List<QuestionListRspVO> result) {
        List<Integer> qids = result.stream().map(QuestionListRspVO::getQuestionId).collect(Collectors.toList());
        QueryWrapper<RoomQuestionLkEntity> lkw = new QueryWrapper<>();

        lkw.eq(ROOM_ID, req.getFromRoomId());

        lkw.in(QUESTION_ID, qids);
        List<RoomQuestionLkEntity> lkList = roomQuestionLkMapper.selectList(lkw);
        Map<Integer, RoomQuestionLkEntity> map = new HashMap<>(16);
        for (RoomQuestionLkEntity entity : lkList) {
            map.put(entity.getQuestionId(), entity);
        }
        for (QuestionListRspVO vo : result) {
            // 补充回答信息, 判断是否有回答
            String key = RedisKey.QUESTION_ANSWERED_LOCK + ":" + req.getRoomId() + ":" + vo.getQuestionId() + ":" + req
                    .getAccountId();
            int a = Boolean.TRUE.equals(redisTemplate.hasKey(key)) ? 1 : 0;
            RoomQuestionLkEntity lk = map.get(vo.getQuestionId());
            vo.setPublish(null == lk ? 0 : lk.getPublish());
            vo.setBind(null == lk ? 0 : lk.getBind());
            vo.setAnswer(a);
        }
    }


    @Override
    public QuestionPage<QuestionListRspVO> watcherList(QuestionListQuery req) {
        String roomId = req.getRoomId();
        // 根据请求参数获取缓存key
        String cacheKey = questionCacheService.getWatchSideHashKey(req);
        // 缓存模式：获取直播间下所有已发布的问卷列表
        Page<QuestionListDTO> cachedData = questionCacheService.getCachedData(roomId, cacheKey);
        if (Objects.isNull(cachedData)) {
            String lockValue = "1";
            if (!RedisLocker.tryLock(cacheKey, lockValue, 10 * 1000L)) {
                try {
                    Thread.sleep(50L);
                } catch (InterruptedException e) {
                    log.error("Interrupted!", e);
                    Thread.currentThread().interrupt();
                }
                return watcherList(req);
            }
            try {
                Page<QuestionListDTO> page = new Page<>(req.getPage(), req.getPageSize());
                QueryWrapper<QuestionListDTO> wrapper = new QueryWrapper<>();
                wrapper.eq("lk.room_id", roomId).eq(Objects.nonNull(req.getPublish()), "lk.publish", req.getPublish());
                // 模糊过滤
                wrapper.and(StringUtils.isNoneEmpty(req.getKeyword()),
                        wq -> wq.like("q.question_id", req.getKeyword()).or().like("q.title", req.getKeyword()));
                wrapper.orderByDesc("lk.question_id");
                cachedData = roomQuestionLkMapper.selectQuestionDetailInRoomPage(page, wrapper);
                if (Objects.nonNull(cachedData)) {
                    questionCacheService.cacheWatchSideQuestionData(roomId, cacheKey, cachedData);
                }
                if (Objects.isNull(cachedData) || cachedData.getTotal() == 0) {
                    return new QuestionPage<>(req.getPage(), req.getPageSize(), 0, Collections.emptyList());
                }
            } finally {
                RedisLocker.unlock(cacheKey, lockValue);
            }
        }

        RoomsEntityDTO roomInfo = roomInnerService.getRoomInfo(roomId);
        String roomSubject = null;
        if (Objects.nonNull(roomInfo)) {
            roomSubject = roomInfo.getSubject();
        }
        List<QuestionListRspVO> result = questionAssembler.convertTos(cachedData.getRecords(), roomSubject);

        // 3、查询问卷的填写人数。
        Map<String, Integer> mapOfCount = new HashMap<>(16);
        Map<String, Integer> mapOfAccount = new HashMap<>(16);
        Map<String, RoomsEntityDTO> mapOfRoomInfo = new HashMap<>(16);
        List<Integer> questionIds = result.stream().map(QuestionListRspVO::getQuestionId).collect(Collectors.toList());
        if (CollectionUtils.isNotEmpty(questionIds)) {
            // 3、查询问卷的填写人数。
            List<QuestionCountVo> qaNumberCounts = questionCacheService.listCachedQuestionAnswerNumber(roomId, questionIds);
            if (CollectionUtils.isEmpty(qaNumberCounts)) {
                String qaNumberCacheKey = questionCacheService.getQuestionAnswerNumberCacheKey(roomId);
                if (!RedisLocker.tryLock(qaNumberCacheKey, "1", 10 * 1000L)) {
                    try {
                        Thread.sleep(50L);
                    } catch (InterruptedException e) {
                        log.error("Interrupted!", e);
                        Thread.currentThread().interrupt();
                    }
                    return watcherList(req);
                }
                try {
                    qaNumberCounts = questionAnswersMapper.getAnswerCountByQuestionIds(questionIds);
                    List<QuestionAnswerNumberCacheDTO> cacheList = questionAssembler.convertToAnswerNumberCaches(qaNumberCounts);
                    // 数据为空时，必须缓存空数据
                    if (CollectionUtils.isEmpty(qaNumberCounts)) {
                        questionIds.forEach(questionId -> cacheList.add(new QuestionAnswerNumberCacheDTO(questionId, 0)));
                    }
                    questionCacheService.cacheQuestionAnswerNumber(roomId, cacheList);
                    log.info("缓存question answer number完成，roomId={}, questionIds={}", roomId, questionIds);
                } finally {
                    RedisLocker.unlock(qaNumberCacheKey, "1");
                }
            }
            for (QuestionCountVo v : qaNumberCounts) {
                mapOfCount.put(String.valueOf(v.getQuestionId()), v.getCount());
            }

            // 4、查询当前用户是否填写了问卷。
            Integer accountId = req.getAccountId();
            List<QuestionCountVo> accountQaRefCounts = questionCacheService.listCachedQuestionAnswerRef(roomId, accountId, questionIds);
            if (CollectionUtils.isEmpty(accountQaRefCounts)) {
                String accountQaRefCacheKey = questionCacheService.getUserQuestionAnswerRefCacheKey(roomId) + ":lock:" + accountId;
                if (!RedisLocker.tryLock(accountQaRefCacheKey, "1", 10 * 1000L)) {
                    try {
                        Thread.sleep(50L);
                    } catch (InterruptedException e) {
                        log.error("Interrupted!", e);
                        Thread.currentThread().interrupt();
                    }
                    return watcherList(req);
                }
                try {
                    List<Integer> questionsIdList = questionAnswersMapper.selectQuestionIdsByRoomIdAndAccountId(roomId, accountId);
                    accountQaRefCounts = questionAnswersMapper.getWhetherMap(CollectionUtils.isEmpty(questionsIdList) ? questionIds : questionsIdList, accountId);
                    questionCacheService.cacheUserQuestionAnswerRef(roomId, accountId, questionAssembler
                            .convertToQuestionAnswerRefCaches(accountQaRefCounts));

                    log.info("缓存question answer ref完成，roomId={}, questionIds={}", roomId, questionIds);
                } finally {
                    RedisLocker.unlock(accountQaRefCacheKey, "1");
                }
            }
            for (QuestionCountVo v : accountQaRefCounts) {
                mapOfAccount.put(String.valueOf(v.getQuestionId()), v.getAnswerId());
            }
        }

        // 给集合中的ilId赋值。
        buildQuestionRoomInfoMap(result, mapOfRoomInfo);
        // 5、给集合中的数据赋值。
        buildQuestionListRspVOList(result, mapOfCount, mapOfAccount, mapOfRoomInfo);
        return new QuestionPage<>(req.getPage(), req.getPageSize(), (int) cachedData.getTotal(), result);
    }

    private void buildQuestionRoomInfoMap(List<QuestionListRspVO> result, Map<String, RoomsEntityDTO> mapOfRoomInfo) {
        List<String> roomIds = new ArrayList<>();
        for (QuestionListRspVO qpl : result) {
            if (qpl.getRoomId() != null && StringUtils.isNotBlank(qpl.getRoomId())) {
                roomIds.add(qpl.getRoomId());
            }
        }
        if (CollectionUtils.isNotEmpty(roomIds)) {
            List<RoomsEntityDTO> rooms = roomInnerService.getRoomInfoByRoomIds(roomIds);
            if (CollectionUtils.isNotEmpty(rooms)) {
                for (RoomsEntityDTO re : rooms) {
                    mapOfRoomInfo.put(re.getRoomId(), re);
                }
            }
        }
    }

    /**
     * 构建问卷结果 list
     *
     * @param result        问卷结果 list
     * @param mapOfCount    查询问卷的填写人数
     * @param mapOfAccount
     * @param mapOfRoomInfo
     */
    private void buildQuestionListRspVOList(List<QuestionListRspVO> result, Map<String, Integer> mapOfCount, Map<String, Integer> mapOfAccount, Map<String, RoomsEntityDTO> mapOfRoomInfo) {
        for (QuestionListRspVO qpl : result) {
            qpl.setAnswererCount(mapOfCount.getOrDefault(String.valueOf(qpl.getQuestionId()), 0));
            if (mapOfAccount.containsKey(String.valueOf(qpl.getQuestionId()))) {
                qpl.setIsFill(1);
                qpl.setAnswerId(mapOfAccount.get(String.valueOf(qpl.getQuestionId())));
            } else {
                qpl.setIsFill(0);
                qpl.setAnswerId(0);
            }
            if (mapOfRoomInfo.containsKey(qpl.getRoomId())) {
                qpl.setIlId(String.valueOf(mapOfRoomInfo.get(qpl.getRoomId()).getIlId()));
            }
        }
    }


    @Override
    public QuestionsEntity create(QuestionCreateReqVO req) {
        Integer accountId = req.getAccountId();
        RoomsEntityDTO room = new RoomsEntityDTO();
        if (StringUtils.isNotBlank(req.getRoomId())) {
            VssToken token = VssTokenThreadLocal.getToken();
            RoomJoinsEntityDTO roomJoins = roomInnerService.getRoomJoins(token.getThirdPartyUserId(), req.getRoomId());
            if (roomJoins == null){
                throw new BusinessException(BizErrorCode.COMP_QUESTION_NOT_PERMISSION);
            }
            room = roomInnerService.getRoomInfo(req.getRoomId());
            accountId = room.getAccountId();
        }
        QuestionsEntity qeOfDb = questionsMapper.selectById(req.getQuestionId());
        QuestionsEntity entity = new QuestionsEntity();
        entity.setQuestionId(req.getQuestionId());
        entity.setTitle(req.getTitle());
        entity.setCover(CharSequenceUtil.nullToEmpty(req.getCover()));
        entity.setDescription(CharSequenceUtil.nullToEmpty(req.getDescription()));
        entity.setExtend(req.getExtend());
        entity.setAccountId(accountId.toString());
        entity.setIsPublic(req.getIsPublic());
        entity.setAppId(req.getAppId());
        entity.setUpdatedAt(LocalDateTime.now());
        try {
            if (qeOfDb != null) {
                // 更新操作
                questionsMapper.updateById(entity);
            } else {
                // 新增操作
                entity.setCreatedAt(LocalDateTime.now());
                questionsMapper.insert(entity);
            }
            if (StringUtils.isNotEmpty(req.getRoomId())) {
                bindRoom(entity.getQuestionId(), Integer.parseInt(entity.getAccountId()), req.getRoomId(), null);

            }
        } finally {
            // 清空房间缓存的问卷信息
            questionCacheService.deleteCachedQuestionDetailData(req.getRoomId());
        }
        return entity;
    }

    @Override
    public Boolean delete(QuestionDeleteReqVO req) {
        // 先查询出来判断下是否被发布了。
        QueryWrapper<QuestionsEntity> where = new QueryWrapper<>();
        where.eq(QUESTION_ID, req.getQuestionId());

//        where.eq(ACCOUNT_ID, req.getAccountId());
        where.last(LIMIT_1);
        QuestionsEntity entity = questionsMapper.selectOne(where);
        if (null == entity) {
            return false;
        }
        // 检查问卷是否发布，已发布的问卷不能删除
        Integer id = roomQuestionLkMapper.questionIsPublish(req.getQuestionId());
        if (null != id) {
            return false;
        }
        // 还没发布，可以删除。
        QuestionsEntity questionsEntity = new QuestionsEntity();
        questionsEntity.setQuestionId(req.getQuestionId());
        questionsEntity.setDeletedAt(LocalDateTime.now());
        int effect = questionsMapper.updateById(questionsEntity);
        return effect == 1;
    }

    /**
     * 检查是否提交过问卷
     *
     * @param roomId
     * @param questionId
     * @return
     */
    @Override
    public Boolean checkSurvey(Integer accountId, String roomId, Integer questionId) {
        RoomJoinsEntityDTO joinsEntity = roomInnerService.getRoomJoinInfo(accountId, roomId);
        if (null == joinsEntity) {
            throw new BusinessException(BizErrorCode.EMPTY_ROOM);
        }
        QueryWrapper<QuestionsEntity> whereOfQ = new QueryWrapper<>();
        whereOfQ.eq(QUESTION_ID, questionId);
        whereOfQ.last(LIMIT_1);
        QuestionsEntity entity = questionsMapper.selectOne(whereOfQ);
        if (null == entity) {
            throw new BusinessException(BizErrorCode.EMPTY_QUESTION);
        }
        QueryWrapper<QuestionAnswersEntity> where = new QueryWrapper<>();
        where.eq(QUESTION_ID, questionId);
        where.eq(ROOM_ID, roomId);
        where.eq(JOIN_ID, joinsEntity.getJoinId());
        where.last(LIMIT_1);
        Integer count = answersMapper.selectCount(where);
        return count > 0;

    }

    @Override
    public List<QuestionSubmitInfoRspVo> submitInfo(String roomId, Integer questionId) {
        int offset = 0;
        int size = 1000;

        return answersMapper.selectSubmitInfo(roomId, questionId, offset, size);
    }

    @Override
    public boolean answer(Integer accountId, String roomId, Integer questionId, Integer answerId, String extend) {
        RoomJoinsEntityDTO joinsEntity = roomInnerService.getRoomJoins(accountId, roomId);
        if (null == joinsEntity) {
            throw new BusinessException(BizErrorCode.COMP_QUESTION_NOT_PERMISSION);
        }
        String key = RedisKey.QUESTION_ANSWERED_LOCK + ":" + roomId + ":" + questionId + ":" + accountId;
        boolean lock = RedisLocker.tryLock(key, "1", 86400 * 1000L);
        if (!lock) {
            throw new BusinessException(BizErrorCode.BIZ_SUBMIT_FAILED);
        }
        Integer joinId = joinsEntity.getJoinId();
        String skey = RedisKey.QUESTION_ANSWERED_LOCK + roomId + ":" + questionId;
        redisTemplate.opsForSet().add(skey, accountId);
        // 添加 房间编号到 set
        redisTemplate.opsForSet().add(RedisKey.FORM_QUEUE_ROOM_IDS, roomId);
        // 游客信息列表
        String vkey = RedisKey.QUEUE_QUESTION_SYNC_VISITOR_INFO + roomId;
        JSONObject obj = new JSONObject();
        obj.put(ROOM_ID, roomId);
        obj.put(ACCOUNT_ID, accountId);
        obj.put(JOIN_ID, joinId);
        obj.put("answer_id", answerId);
        obj.put("extend", extend);
        obj.put(QUESTION_ID, questionId);
        pushToRedis(vkey, obj);
        QuestionAnswersEntity entity = new QuestionAnswersEntity();
        entity.setAnswerId(answerId);
        entity.setRoomId(roomId);
        entity.setQuestionId(questionId);
        entity.setExtend(extend);
        entity.setAccountId(accountId);
        entity.setJoinId(joinId);
        answersMapper.insert(entity);
        //清除用户答卷QA关系缓存
        questionCacheService.deleteCachedQuestionAnswerRef(roomId, String.valueOf(accountId));
        return true;
    }

    @Override
    public boolean visitorAnswer(Integer accountId, String roomId, Integer questionId,
                                 Integer answerId, String extend, String deviceId) {
        String key = RedisKey.QUESTION_ANSWERED_LOCK + ":" + roomId + ":" + questionId + ":" + deviceId;
        boolean lock = RedisLocker.tryLock(key, "1", 86400 * 1000L);
        if (!lock) {
            throw new BusinessException(BizErrorCode.BIZ_SUBMIT_FAILED);
        }
        JSONObject obj = new JSONObject();
        obj.put(ROOM_ID, roomId);
        obj.put(ACCOUNT_ID, accountId);
        obj.put(JOIN_ID, 0);
        obj.put("answer_id", answerId);
        obj.put("extend", "{}");
        obj.put(QUESTION_ID, questionId);
        // 房间, 问卷验证
        QuestionsEntity question = questionsMapper.selectById(questionId);
        if (null == question) {
            throw new BusinessException(BizErrorCode.EMPTY_QUESTION);
        }
        RoomsEntityDTO room = roomInnerService.getRoomInfo(roomId);
        if (null == room) {
            throw new BusinessException(BizErrorCode.EMPTY_QUESTION);
        }
        if (accountId > 0) {
            String key2 = RedisKey.QUESTION_ANSWERED_LOCK + ":" + roomId + ":" + questionId + ":" + accountId;
            boolean lock2 = RedisLocker.tryLock(key2, "1", 86400 * 1000L);
            if (!lock2) {
                throw new BusinessException(BizErrorCode.BIZ_SUBMIT_FAILED);
            }
            // 参数用户
            RoomJoinsEntityDTO joiner = roomInnerService.getRoomJoins(accountId, roomId);
            if (null != joiner) {
                obj.put(JOIN_ID, joiner.getJoinId());
                obj.put(ACCOUNT_ID, accountId);
            }
        }
        // 添加 房间编号到 set
        redisTemplate.opsForSet().add(RedisKey.FORM_QUEUE_ROOM_IDS, roomId);
        // 游客信息列表
        String vkey = RedisKey.QUEUE_QUESTION_SYNC_VISITOR_INFO + roomId;
        pushToRedis(vkey, obj);
        //未登录用户提交问卷记录 入库
        QuestionAnswersEntity entity = new QuestionAnswersEntity();
        entity.setAnswerId(answerId);
        entity.setRoomId(roomId);
        entity.setQuestionId(questionId);
        entity.setExtend(extend);
        entity.setAccountId(accountId);
        entity.setJoinId(0);
        answersMapper.insert(entity);
        return true;
    }

    @Override
    public boolean publishCancel(Integer accountId, String roomId, Integer questionId) {
        QueryWrapper<QuestionsEntity> where = new QueryWrapper<>();
        where.eq(QUESTION_ID, questionId);
        where.last(LIMIT_1);
        QuestionsEntity question = questionsMapper.selectOne(where);
        if (null == question) {
            throw new BusinessException(BizErrorCode.EMPTY_QUESTION);
        }
        RoomJoinsEntityDTO joiner = roomInnerService.getRoomJoins(accountId, roomId);
        if (joiner.getRoleName().equals(QuestionConsts.ROLE_AUDIENCE)) {
            throw new BusinessException(BizErrorCode.COMP_QUESTION_NOT_PERMISSION);
        }
        UpdateWrapper<RoomQuestionLkEntity> up = new UpdateWrapper<>();
        up.eq(ROOM_ID, roomId);
        up.eq(QUESTION_ID, question.getQuestionId());
        RoomQuestionLkEntity entity = new RoomQuestionLkEntity();
        entity.setPublish(0);
        entity.setUpdatedAt(LocalDateTime.now());
        int effect = roomQuestionLkMapper.update(entity, up);
        return effect == 1;
    }

    @Override
    public boolean publish(Integer accountId, String roomId, Integer questionId) {
        questionPushIntervalCheck(roomId);
        String questionRePushIntervalKey = RedisKey.QUESTION_REPUSH_INTERVAL_KEY + roomId;
        // 根据roomId, 获取参会用户信息
        RoomJoinsEntityDTO entity = roomInnerService.getRoomJoins(accountId, roomId);
        if (null == entity) {
            throw new BusinessException(BizErrorCode.COMP_QUESTION_NOT_PERMISSION);
        }
        // 这个接口是在直播间都可以调用所以这里需要判断调用接口的用户的角色进而判断是否可以发布.
        // 参数用户
        if (QuestionConsts.ROLE_AUDIENCE.equals(entity.getRoleName())) {
            throw new BusinessException(BizErrorCode.COMP_QUESTION_NOT_PERMISSION);
        }
        int role = 1;
        try {
            role = Integer.parseInt(entity.getRoleName());
        } catch (Exception e) {
            log.error("未能正确获取到当前用户的role_name", e);
        }
        RoomsEntityDTO roomInfo = roomInnerService.getRoomInfo(roomId);
        boolean result = doPublish(roomInfo.getAccountId(), entity.getNickname(), roomId, questionId, role);
        redisTemplate.opsForValue().set(questionRePushIntervalKey, roomId, QUESTION_RE_PUSH_INTERVAL_TIMEOUT, TimeUnit.SECONDS);
        return result;

    }

    @Override
    public boolean repush(Integer accountId, String roomId, Integer questionId) {
        questionPushIntervalCheck(roomId);
        String questionRePushIntervalKey = RedisKey.QUESTION_REPUSH_INTERVAL_KEY + roomId;
        try {
            RoomsEntityDTO room = roomInnerService.getRoomInfo(roomId);
            if (null == room) {
                throw new BusinessException(BizErrorCode.EMPTY_ROOM);
            }
            // 发布问卷, 问卷信息本地静态化
            String ossUrl = writeInfoLocal(questionId);
            //未答卷用户信息分批发送防止消息body体过长
            boolean flag = true;
            String channel = room.getChannelId();
            String type = "questionnaire_repush_guoxin";
            JSONArray arr = new JSONArray();

            //pass查询人数
            int onlinePageSize = 1000;

            Long onlineCount = paasService.getUserOnlineCount(room.getChannelId());
            int totalPage = (int) Math.ceil(onlineCount * 1.0 / onlinePageSize);

            List<Integer> resultIds = new ArrayList<>();
            resultIds = async(totalPage, 1, onlinePageSize, room, resultIds);

            if (resultIds.isEmpty()) {
                return true;
            }

            Integer page = 1;
            Integer offset;
            Integer pageSize = 1000;
            while (flag) {
                offset = (page - 1) * pageSize;
                log.info("开始查询页码=" + page);
                log.info("开始查询 offset=" + offset);
                // 这里不要看方法名字，实际执行的SQL查询的是 已经填写过问卷的用户信息
                List<QuestionUnAnswerJoiner> joinerList =
                        answersMapper.selectUnAnswerJoiner(questionId, roomId, offset, pageSize, resultIds);
                page++;
                if (joinerList.size() < pageSize) {
                    flag = false;
                }
                for (QuestionUnAnswerJoiner joiner : joinerList) {
                    if (joiner.getAccountId() != null) {
                        arr.add(joiner.getAccountId());
                    }
                }

            }

            String idsUrl = writeIdsLocal(questionId, arr);

            JSONObject msg = new JSONObject();
            msg.put("type", type);
            msg.put("questionnaire_id", questionId);
            msg.put(ACCOUNT_ID, accountId);
            msg.put("info_url", ossUrl);
            msg.put("ids_url", idsUrl);
            paasService.sendMessage(channel, JsonUtil.toJsonString(msg));
            redisTemplate.opsForValue().set(questionRePushIntervalKey, roomId, QUESTION_RE_PUSH_INTERVAL_TIMEOUT, TimeUnit.SECONDS);
        } catch (BusinessException ex) {
            throw ex;
        } catch (Exception ex) {
            log.error("Exception", ex);
            throw new BusinessException(BizErrorCode.COMP_QUESTION_PUSH_FAILED);
        }
        return true;
    }

    /**
     * 问卷推屏时间间隔检查
     *
     * @param roomId 房间id
     * @return 校验的 Key
     */
    private void questionPushIntervalCheck(String roomId) {
        String questionRePushIntervalKey = RedisKey.QUESTION_REPUSH_INTERVAL_KEY + roomId;
//        设置问卷推屏限频,一个房间的问卷只有间隔一定时间才能进行推屏
        Object rePushInterval = redisTemplate.opsForValue().get(questionRePushIntervalKey);
        if (Objects.nonNull(rePushInterval)) {
            throw new BusinessException(BizErrorCode.BIZ_ACT_FREQUENT);
        }
    }


    private List<Integer> async(
            int totalPage, int startIndex, int onlinePageSize,
            RoomsEntityDTO room, List<Integer> accountIds) {
        List<CompletableFuture<List<Integer>>> futures = new ArrayList<>();
        String channelId = room.getChannelId();
        for (int i = startIndex; i <= totalPage; i++) {
            final int j = i;

            CompletableFuture<List<Integer>> future = CompletableFuture.supplyAsync(() ->
                    watcher(channelId, j, onlinePageSize)
            );
            futures.add(future);
            if (i == totalPage) {
                Iterator<CompletableFuture<List<Integer>>> iterator = futures.iterator();
                while (iterator.hasNext()) {
                    CompletableFuture<List<Integer>> result = iterator.next();
                    try {
                        accountIds.addAll(result.get());
                    } catch (InterruptedException e) {
                        log.error("获取数据失败", e);
                        Thread.currentThread().interrupt();
                        throw new BusinessException(GlobalErrorCode.FAILED);
                    } catch (ExecutionException e) {
                        log.error("获取数据失败", e);
                        throw new BusinessException(GlobalErrorCode.FAILED);
                    }
                    iterator.remove();
                }

            }
        }
        return accountIds;
    }


    private List<Integer> watcher(String channelId, int onlinePage, int onlinePageSize) {
        List<Integer> ids = paasService.getUserIdList(channelId, onlinePage, onlinePageSize);

        if (CollectionUtils.isEmpty(ids)) {
            return Collections.emptyList();
        }
        return ids;
    }


    @Override
    public QuestionsEntity copy(Integer accountId, String roomId, Integer questionId) {
        String appId = paasService.getAppId();
        if (StringUtils.isNotEmpty(roomId)) {
            RoomsEntityDTO room = roomInnerService.getRoomInfo(roomId);
            VssToken token = VssTokenThreadLocal.getToken();
            RoomJoinsEntityDTO roomJoins = roomInnerService.getRoomJoins(token.getThirdPartyUserId(), roomId);
            if (roomJoins == null){
                throw new BusinessException(BizErrorCode.COMP_QUESTION_NOT_PERMISSION);
            }
        }
        QuestionsEntity question = questionsMapper.selectById(questionId);
        if (null == question) {
            throw new BusinessException(BizErrorCode.EMPTY_QUESTION);
        }
        JSONObject doc = paasService.getFormInfo(questionId.toString());
        Map<String, Object> param = new HashMap<>(16);
        param.put("title", doc.getString("title"));
        param.put("description", doc.getString("description"));
        param.put("imgUrl", doc.getString("imgUrl"));
        param.put("publish", "Y");
        param.put("owner_id", accountId);
        param.put("detail", formatDetail(doc.getJSONArray("detail")));
        JSONObject object = paasService.createForm(param);
        if (object == null || !object.containsKey("id")) {
            throw new BusinessException(BizErrorCode.EMPTY_QUESTION);
        }
        LocalDateTime now = LocalDateTime.now();
        QuestionsEntity entity = new QuestionsEntity();
        BeanUtils.copyProperties(question, entity);
        entity.setQuestionId(object.getInteger("id"));
        entity.setAppId(appId);
        entity.setCreatedAt(now);
        entity.setUpdatedAt(now);
        try {
//            entity.setAccountId(accountId.toString());
            questionsMapper.insert(entity);
            bindRoom(entity.getQuestionId(), Integer.parseInt(accountId.toString()), roomId, null);
        } finally {
            // 清空房间缓存的问卷信息
            questionCacheService.deleteCachedQuestionDetailData(roomId);
        }
        return entity;
    }

    @Override
    public QuestionsEntity info(QuestionInfoReqVo reqVo) {
        QuestionsEntity questions;
        if (reqVo.getQuestionId() != null) {
            //根据questionId查找
            Object v = redisTemplate.opsForValue().get(RedisKey.QUESTION_INFO_OP_KEY + reqVo.getQuestionId());
            if (v == null) {
                questions = questionsMapper.selectOne(Wrappers.<QuestionsEntity>lambdaQuery()
                        .eq(QuestionsEntity::getQuestionId, reqVo.getQuestionId())
                        .isNull(QuestionsEntity::getDeletedAt));
                Optional.ofNullable(questions).orElseThrow(() -> new BusinessException(BizErrorCode.EMPTY_QUESTION));
                redisTemplate.opsForValue().set(RedisKey.QUESTION_INFO_OP_KEY + reqVo.getQuestionId(), JsonUtil.toJsonString(questions));
            } else {
                questions = JsonUtil.objectFromJSONString(v.toString(), QuestionsEntity.class);
            }
        } else {
            //根据sourceId查找
            Object v = redisTemplate.opsForValue().get(RedisKey.QUESTION_INFO_BY_SOURCE_ID + reqVo.getSourceId());
            if (v == null) {
                questions = questionsMapper.selectOne(Wrappers.<QuestionsEntity>lambdaQuery()
                        .eq(QuestionsEntity::getSourceId, reqVo.getSourceId())
                        .isNull(QuestionsEntity::getDeletedAt));
                Optional.ofNullable(questions).orElseThrow(() -> new BusinessException(BizErrorCode.EMPTY_QUESTION));
                redisTemplate.opsForValue().set(RedisKey.QUESTION_INFO_BY_SOURCE_ID + reqVo.getSourceId(), JsonUtil.toJsonString(questions));
            } else {
                questions = JsonUtil.objectFromJSONString(v.toString(), QuestionsEntity.class);
            }
        }
        Optional.ofNullable(questions).orElseThrow(() -> new BusinessException(BizErrorCode.EMPTY_QUESTION));
        if (!questions.getAccountId().equals(reqVo.getAccountId())) {
            throw new BusinessException(BizErrorCode.EMPTY_QUESTION);
        }
        return questions;
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public Integer unbindRoom(Integer questionId, String roomId, Integer accountId) {
        QuestionsEntity questions;
        //获取question
        Object v = redisTemplate.opsForValue().get(RedisKey.QUESTION_INFO_BY_ID + questionId);
        if (v == null) {
            questions = questionsMapper.selectOne(
                    Wrappers.<QuestionsEntity>lambdaQuery().eq(QuestionsEntity::getQuestionId, questionId)
                            .isNull(QuestionsEntity::getDeletedAt));
            if (questions == null) {
                throw new BusinessException(BizErrorCode.EMPTY_QUESTION);
            }
            redisTemplate.opsForValue()
                    .set(RedisKey.QUESTION_INFO_BY_ID + questionId, JsonUtil.toJsonString(questions), 1, TimeUnit.DAYS);
        } else {
            questions = JsonUtil.objectFromJSONString(v.toString(), QuestionsEntity.class);
        }
        if (questions == null) {
            throw new BusinessException(BizErrorCode.EMPTY_QUESTION);
        }
        if (!questions.getAccountId().equals(String.valueOf(accountId))) {
            throw new BusinessException(BizErrorCode.COMP_QUESTION_NO_DELETE_ABILITY);
        }
        //获取roomQuestionLk
        RoomQuestionLkEntity roomQuestionLk;
        Object s = redisTemplate.opsForValue().get(RedisKey.QUESTION_INFO_KEY + roomId + "and" + questionId);
        if (s == null) {
            roomQuestionLk = roomQuestionLkMapper.selectOne(
                    Wrappers.<RoomQuestionLkEntity>lambdaQuery().eq(RoomQuestionLkEntity::getRoomId, roomId)
                            .eq(RoomQuestionLkEntity::getQuestionId, questionId));
            if (roomQuestionLk == null) {
                throw new BusinessException(BizErrorCode.EMPTY_QUESTION);
            }
            redisTemplate.opsForValue()
                    .set(RedisKey.QUESTION_INFO_KEY + roomId + "and" + questionId, JsonUtil.toJsonString(roomQuestionLk), 1,
                            TimeUnit.DAYS);
        } else {
            roomQuestionLk = JsonUtil.objectFromJSONString(s.toString(), RoomQuestionLkEntity.class);
        }
        if (roomQuestionLk == null) {
            throw new BusinessException(BizErrorCode.EMPTY_QUESTION);
        }
        // 已发布的问卷不能解绑
        if (roomQuestionLk.getPublish() == 1) {
            throw new BusinessException(BizErrorCode.COMP_QUESTION_NOT_DELETE);
        }
        //强制删除
        return roomQuestionLkMapper.deleteById(roomQuestionLk.getId());
    }

    @Override
    public QuestionsEntity update(QuestionUpdateReqVo req) {
        QueryWrapper<QuestionsEntity> where = new QueryWrapper<>();
        where.eq(QUESTION_ID, req.getQuestionId());
//        where.eq(ACCOUNT_ID, req.getAccountId());

        where.isNull(DELETED_AT);
        where.last(LIMIT_1);
        QuestionsEntity question = questionsMapper.selectOne(where);
        if (null == question) {
            throw new BusinessException(BizErrorCode.EMPTY_QUESTION);
        }
        // 已发布的问卷不能编辑
        Integer id = roomQuestionLkMapper.questionIsPublish(question.getQuestionId());
        if (null != id) {
            // 已经发布过
            throw new BusinessException(BizErrorCode.COMP_QUESTION_NOT_EDIT);
        }
        try {
            QuestionsEntity upe = new QuestionsEntity();
            upe.setQuestionId(req.getQuestionId());
            upe.setUpdatedAt(LocalDateTime.now());
            upe.setTitle(req.getTitle());
            upe.setCover(req.getCover());
            upe.setDescription(req.getDescription());
            upe.setExtend(req.getExtend());
            questionsMapper.updateById(upe);

            redisTemplate.delete(RedisKey.QUESTION_INFO_OSS_URL + req.getQuestionId().toString());

            // 7、更新绑定的room的操作。
            LambdaQueryWrapper<RoomQuestionLkEntity> whereOfRqle = Wrappers.lambdaQuery();
            whereOfRqle.eq(RoomQuestionLkEntity::getQuestionId, req.getQuestionId())
                    .last(LIMIT_1);
            RoomQuestionLkEntity rqOfOld = roomQuestionLkMapper.selectOne(whereOfRqle);
            if (rqOfOld == null) {
                // 执行绑定操作
                bindRoom(req.getQuestionId(), req.getAccountId(), req.getRoomId(), null);
            } else if (!rqOfOld.getRoomId().equals(req.getRoomId())) {

                // 执行更新操作。更换绑定房间
                LambdaUpdateWrapper<RoomQuestionLkEntity> whereOfLk = Wrappers.lambdaUpdate();
                whereOfLk.eq(RoomQuestionLkEntity::getQuestionId, req.getQuestionId());
                whereOfLk.eq(RoomQuestionLkEntity::getRoomId, req.getRoomId());
                RoomQuestionLkEntity lkParam = new RoomQuestionLkEntity();
                lkParam.setBind(1);
                lkParam.setRoomId(req.getRoomId());
                lkParam.setUpdatedAt(LocalDateTime.now());
                roomQuestionLkMapper.update(lkParam, whereOfLk);
            }
            // 8、删除缓存。
            if (rqOfOld != null) {
                redisTemplate.delete(RedisKey.QUESTION_INFO_KEY + rqOfOld.getRoomId() + "and" + req.getQuestionId());
            }
        } finally {
            // 清空房间缓存的问卷信息
            questionCacheService.deleteCachedQuestionDetailData(req.getRoomId());
        }
        return question;
    }

    @Override
    public QuestionListStatRspVO<QuestionStatDTO> statList(QuestionListQuery query) {
        Integer offset = (query.getPage() - 1) * query.getPageSize();
        Integer count = questionAnswersMapper.countGroupByRoomIdAndQuestionId(query.getAccountId(),
                query.getBeginDate(), query.getEndDate());
        List<QuestionStatDTO> list = new ArrayList<>();
        if (count > 0) {
            list = questionAnswersMapper.statGroupByRoomIdAndQuestionId(query.getAccountId(),
                    query.getBeginDate(), query.getEndDate(), offset, query.getPageSize());
        }
        QuestionListStatRspVO<QuestionStatDTO> result = new QuestionListStatRspVO<>();
        result.setList(list);
        result.setPage(query.getPage());
        result.setTotal(count);
        result.setPagesize(query.getPageSize());
        return result;
    }

    @Override
    public QuestionNumRspVO statNum(QuestionNumReqVO query) {
        QuestionNumRspVO result = new QuestionNumRspVO();
        Integer roomNum = roomQuestionLkMapper.countQuestionNum(query.getRoomId(), query.getAccountId(),
                query.getBeginDate(), query.getEndDate());
        result.setByRoomNum(roomNum);
        Integer answerNum = questionAnswersMapper.countAnswerNum(query.getRoomId(), query.getAccountId(),
                query.getBeginDate(), query.getEndDate());
        result.setByAccountNum(answerNum);
        return result;
    }

    /**
     * 查询某个房间再开播后的参与问答的人
     *
     * @param roomId
     * @param startTime
     * @param page
     * @param pageSize
     * @param queryCount 是否查询所有数据
     * @return
     */
    @Override
    public QuestionPageRspVO<Integer> queryQuesAccounts(String roomId, String startTime, int page, int pageSize, boolean queryCount) {
        int totalCount = 0;
        if (queryCount) {
            totalCount = questionAnswersMapper.queryAllPartakeAccountsCount(roomId, startTime);
        }
        List<QuestionAnswersEntity> questionAnswersEntities = questionAnswersMapper.queryAllPartakeAccounts(roomId, startTime, (page - 1) * pageSize, pageSize);
        List<Integer> accountIds = questionAnswersEntities.stream().map(QuestionAnswersEntity::getAccountId).collect(Collectors.toList());
        QuestionPageRspVO<Integer> questionPageRspVO = new QuestionPageRspVO<>();
        questionPageRspVO.setPage(page);
        questionPageRspVO.setTotalCount(totalCount);
        questionPageRspVO.setData(accountIds);
        return questionPageRspVO;
    }

    /**
     * 查询某个房间再开播后的参与问答的总人数
     */
    @Override
    public Integer queryAllPartakeAccountsCount(String roomId, String startTime) {
        return questionAnswersMapper.queryAllPartakeAccountsCount(roomId, startTime);
    }

    @Override
    public IndexDto getCount(String beginTime) {
        return getCount(beginTime, null);
    }

    @Override
    public IndexDto getCount(String beginTime, String orgCode) {
        DateTimeFormatter df = DateTimeFormatter.ofPattern(CommonConstant.DATETIME_FORMAT);
        LocalDateTime time = LocalDateTime.parse(beginTime, df);
        LocalDateTime today = LocalDate.now().atStartOfDay();
        QueryWrapper<QuestionsEntity> queryWrapper = new QueryWrapper<>();
        queryWrapper.isNull(DELETED_AT);
        Integer total = questionsMapper.selectCount(queryWrapper);

        Integer day = questionsMapper.selectCount(queryWrapper.ge(CREATED_AT, time));
        queryWrapper.clear();
        queryWrapper.isNull(DELETED_AT);
        queryWrapper.le(CREATED_AT,today);
        Integer week = questionsMapper.selectCount(queryWrapper.ge(CREATED_AT, time.plusDays(-1 * SelectTimeIntervalOptionsEnum.PAST_7_DAYS.getDay())));
        queryWrapper.clear();
        queryWrapper.isNull(DELETED_AT);
        queryWrapper.le(CREATED_AT,today);
        Integer month = questionsMapper.selectCount(queryWrapper.ge(CREATED_AT, time.plusDays(-1 * SelectTimeIntervalOptionsEnum.PAST_30_DAYS.getDay())));
        queryWrapper.clear();
        queryWrapper.isNull(DELETED_AT);
        queryWrapper.le(CREATED_AT,today);
        Integer year = questionsMapper.selectCount(queryWrapper.ge(CREATED_AT, time.minusYears(1L)));
        IndexDto indexDto = new IndexDto();
        indexDto.setTotal(total);
        indexDto.setDay(day);
        indexDto.setWeek(week);
        indexDto.setMonth(month);
        indexDto.setYear(year);
        return indexDto;
    }

    private String formatDetail(JSONArray detail) {
        JSONArray array = new JSONArray();
        for (int i = 0; i < detail.size(); i++) {
            JSONObject obj = detail.getJSONObject(i);
            obj.remove("id");
            obj.remove("third_party_user_id");
            obj.remove(CREATED_AT);
            obj.remove("app_id");
            array.add(obj);
        }
        return JSON.toJSONString(array);
    }

    private void pushToRedis(String key, JSONObject value) {
        redisTemplate.opsForSet().add(RedisKey.QUEUE_NAMES_KEY, key);
        redisTemplate.opsForList().rightPush(key, value.toString());
    }

}
